package kafka

import (
	"data-center/config"
	"data-center/pkg/queue"
	"fmt"
	"github.com/IBM/sarama"
)

func init() {
	// 注册工厂函数
	queue.RegisterKafkaProducer(NewProducer)
	queue.RegisterKafkaConsumer(NewConsumer)
}

// Producer Kafka生产者
type Producer struct {
	producer sarama.SyncProducer
	topics   map[string]string
}

// Consumer Kafka消费者
type Consumer struct {
	consumer sarama.Consumer
	topics   []string
}

// NewProducer 创建Kafka生产者
func NewProducer(cfg config.QueueConfig) (queue.Producer, error) {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Retry.Max = 5
	config.Producer.Return.Successes = true

	p, err := sarama.NewSyncProducer([]string{cfg.Kafka.URI}, config)
	if err != nil {
		return nil, fmt.Errorf("failed to create Kafka producer: %w", err)
	}

	// 映射队列名称
	topics := map[string]string{
		"task":  cfg.Queues.Task,
		"file":  cfg.Queues.File,
		"index": cfg.Queues.Index,
	}

	return &Producer{
		producer: p,
		topics:   topics,
	}, nil
}

// Publish 发布消息
func (p *Producer) Publish(queueType string, message []byte) error {
	topic, exists := p.topics[queueType]
	if !exists {
		return fmt.Errorf("unknown queue type: %s", queueType)
	}

	msg := &sarama.ProducerMessage{
		Topic: topic,
		Value: sarama.StringEncoder(message),
	}

	_, _, err := p.producer.SendMessage(msg)
	return err
}

// Close 关闭生产者
func (p *Producer) Close() {
	if p.producer != nil {
		p.producer.Close()
	}
}

// NewConsumer 创建Kafka消费者
func NewConsumer(cfg config.QueueConfig, queueTypes []string) (queue.Consumer, error) {
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true

	c, err := sarama.NewConsumer([]string{cfg.Kafka.URI}, config)
	if err != nil {
		return nil, fmt.Errorf("failed to create Kafka consumer: %w", err)
	}

	return &Consumer{
		consumer: c,
		topics:   queueTypes,
	}, nil
}

// Consume 消费消息
func (c *Consumer) Consume(timeout int) ([]byte, error) {
	// 实现消费逻辑
	return nil, fmt.Errorf("not implemented")
}

// Close 关闭消费者
func (c *Consumer) Close() {
	if c.consumer != nil {
		c.consumer.Close()
	}
}
